package com.atbeijing.D08;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import java.util.ArrayList;
import java.util.HashMap;

/**
 *  flink将数据写入es
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        ArrayList<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));

        ElasticsearchSink.Builder<SensorReading> esBuilder = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<SensorReading>() {
                    @Override
                    public void process(SensorReading r, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                        HashMap<String, String> data = new HashMap<>();
                        data.put(r.id, r.temperature + "");

                        IndexRequest indexRequest = Requests
                                .indexRequest()
                                .index("sensor-reading")
                                .type("type") // es6必须写type
                                .source(data);

                        requestIndexer.add(indexRequest);
                    }
                }
        );

        // 设置每次向es写入多少条数据
        esBuilder.setBulkFlushMaxActions(1);

        env
                .addSource(new SensorSource())
                .addSink(esBuilder.build());

        env.execute();
    }
}
